package zk_01.zookeeper;

import java.io.IOException;
import java.util.Random;

import org.apache.zookeeper.CreateMode;
import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.KeeperException.ConnectionLossException;
import org.apache.zookeeper.KeeperException.NodeExistsException;
import org.apache.zookeeper.WatchedEvent;
import org.apache.zookeeper.Watcher;
import org.apache.zookeeper.ZooDefs;
import org.apache.zookeeper.ZooKeeper;
import org.apache.zookeeper.data.Stat;
import org.omg.PortableServer.POAManagerPackage.State;


/*
 * 同步获得管理权，其实就是在ZK中创建一个临时节点，这个临时节点对应的数据（byte数组）是其独有的标识，比如虚拟的进程ID
 * */
public class Master1 implements Watcher{

	private ZooKeeper zk;
	private String hostPort;
	private String serverId;
	private boolean isLeader = false;
	private String leader;
	
	public Master1() {}
	
	public Master1(String hostPort){
		this.hostPort=hostPort;
		Random rd = new Random();
		serverId = Integer.toString(rd.nextInt());
		leader = serverId;
	}
	
	public void startZK() throws IOException{
		zk = new ZooKeeper(hostPort,15000,this);
	}
	
	public void stopZK() throws InterruptedException{
		zk.close();
	}
	
	boolean checkMaster() throws InterruptedException{
		while(true){
			try {
				Stat stat = new Stat();
				byte data[] = zk.getData("/master", false, stat);
				isLeader = new String(data).equals(serverId);
				if(!isLeader){
					leader = new String(data);
				}
				return true;
			} catch (KeeperException e) {//NoNodeException ConnectionLossException都是KeeperException的子类，而且都属于需要重新去判断是否存在主节点，所以返回false;
				return false;
			} 
		}
	}
	
	void runForMaster() throws InterruptedException{
		while(true){
			try{
				zk.create("/master", serverId.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL);
			}catch(NodeExistsException e){
				isLeader = false;
				break;
			}catch (KeeperException e) {
				e.printStackTrace();
			}
			if(checkMaster()) break;
		}
	}
	
	public boolean getIsLeader(){
		return isLeader;
	}
	
	public String getLeader(){
		return leader;
	}
	
	public static void main(String[] args) {
		Master1 m = new Master1("192.168.1.113:2181");
		try {
			m.startZK();
			
			m.runForMaster();
			
			if(m.getIsLeader()){
				System.out.println("I am the Leader");
			}else{
				System.out.println("Someone else is the Leader:"+m.getLeader());
			}
			Thread.sleep(60000);
			
			m.stopZK();
		} catch (IOException e) {
			e.printStackTrace();
		} catch (InterruptedException e) {
			e.printStackTrace();
		}
	}
	
	public void process(WatchedEvent arg0) {
		System.out.println(arg0);
	}

}
